package com.asap.demo.function.sink;

import com.alibaba.fastjson.JSONObject;
import com.asap.demo.utils.FileReader;
import com.asap.rule.StandardEvent;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;

/**
 * @author wangbh
 * @Description: 入es类
 * @date 2021/11/24 17:31
 */
public class DealElasticsearchSinkFunction implements ElasticsearchSinkFunction<StandardEvent> {
    private static final Logger logger = LoggerFactory.getLogger(DealElasticsearchSinkFunction.class);

    String mapData ;
    @Override
    public void open() throws Exception {
        mapData=  new FileReader().readFileNIO("/home/asap/wbh/conf/mapping_7x.properties");
        logger.info("mapDate"+mapData);
    }

    @Override
    public void close() throws Exception {

    }

    @Override
    public void process(StandardEvent standardEvent, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
        //logger.info("indexName:"+standardEvent.getField("indexName")+"|SNOW_ID："+standardEvent.getField("SNOW_ID"));
        String indexName=standardEvent.getField("indexName");
        standardEvent.getAllFields().remove("indexName");
        requestIndexer.add(Requests.indexRequest()
                .index(indexName)
                .source( new JSONObject(standardEvent.getAllFields()).toString().getBytes(Charset.forName("UTF-8")), XContentType.JSON));
    }

}
